package com.gitee.wsl.flow.transform.time

import com.gitee.wsl.concurrent.collections.ConcurrentMutableList
import com.gitee.wsl.ext.base.currentTimeMillis
import com.gitee.wsl.flow.transform.transform
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.produceIn
import kotlinx.coroutines.isActive
import kotlinx.coroutines.withTimeout
import kotlin.time.Duration
import kotlin.time.Duration.Companion.seconds
import kotlin.time.ExperimentalTime
import kotlin.time.TimedValue
import kotlin.time.measureTimedValue

/**
 * Retrieves a [Flow] that emits the items of the source [Flow] but applies a timeout policy for each emitted item. If
 * the next item isn't emitted within the provided timeout [duration] starting from its predecessor, then the resulting
 * [Flow] terminates with a [TimeoutCancellationException] by throwing it as an error.
 *
 * @param [duration] The timeout [Duration].
 *
 * @author chRyNaN
 */
fun <T> Flow<T>.timeout(duration: Duration): Flow<T> = channelFlow {
    val channel = produceIn(this)

    while (isActive) {
        val item = withTimeout(duration) { channel.receive() }

        send(item)
    }
}

/**
 * Retrieves a [Flow] that emits the items of the source [Flow] wrapped in a [TimedValue] containing the elapsed time
 * interval duration between items being emitted.
 *
 * @author chRyNaN
 */
fun <T> Flow<T>.timedValue(): Flow<TimedValue<T>> = channelFlow {
    val channel = produceIn(this)

    while (isActive) {
        val timedValue = measureTimedValue { channel.receive() }

        send(timedValue)
    }
}

/**
 * Delays each emitted item from this [Flow] by the provided [duration].
 */
@ExperimentalTime
fun <T> Flow<T>.delayEach(duration: Duration): Flow<T> = onEach {
    delay(duration)
}

/**
 * take the first data in every [periodMillis] and drop the rest
 */
fun <T> Flow<T>.throttleFirst(periodMillis: Long): Flow<T> {
    var lastTime = 0L
    return transform { upstream ->
        val currentTime = currentTimeMillis
        if (currentTime - lastTime > periodMillis) {
            lastTime = currentTime
            emit(upstream)
        }
    }
}


/**
 * Returns a list of flow items captured for [duration]
 * @param duration of capturing
 * @return a list of flow items captured for [duration]
 */
suspend fun <T> Flow<T>.captureFor(duration: Duration): List<T> {
    val output = ConcurrentMutableList<T>()

    try {
        withTimeout(duration) {
            collect { output += it }
        }
    } catch (e: TimeoutCancellationException) {
        // ignore
    }
    return output
}

/** @return the first element matching [condition] within [timeout]. */
private suspend fun <T> Flow<T>.awaitFirst(message: String, expectedResult: String, timeout: Duration, condition: suspend (T) -> Boolean = { true }): T {
    val output = ConcurrentMutableList<T>()
    return try {
        withTimeout(timeout) {
            onEach { element ->
                output += element
            }.first(condition)
        }
    } catch (e: TimeoutCancellationException) {
        throw Exception(
            listOf(message, expectedResult, "Elements emitted: $output")
                .filter { it.isNotEmpty() }
                .joinToString(separator = ". "),
        )
    }
}

private val defaultTimeout = 5.seconds

/** @return the first element matching [condition] within [timeout]. */
suspend fun <T> Flow<T>.awaitFirst(message: String = "", timeout: Duration = defaultTimeout, condition: suspend (T) -> Boolean = { true }): T =
    awaitFirst(message, "No element matching condition found.", timeout, condition)


/** Asserts that the flow emits an element matching [condition] within [timeout]. */
suspend fun <T> Flow<T>.assertEmits(message: String = "", timeout: Duration = defaultTimeout, condition: suspend (T) -> Boolean) {
    awaitFirst(message, timeout, condition)
}

/** Asserts that the flow emits an element matching [expected] within [timeout]. */
suspend fun <T> Flow<T>.assertEmits(expected: T, message: String = "", timeout: Duration = defaultTimeout) {
    awaitFirst(message, "Expected to emit $expected.", timeout) { it == expected }
}
